#include "pc/peer_connection.h"
#include "pc/session_description.h"
#include "ice/ice_credentials.h"
#include "ice/port_allocator.h"
#include <rtc_base/logging.h>
#include <absl/algorithm/container.h>
#include "pc/stream_params.h"
#include <modules/rtp_rtcp/source/rtp_packet_received.h>
#include <modules/rtp_rtcp/source/rtcp_packet/sender_report.h>
#include <modules/rtp_rtcp/source/rtcp_packet/receiver_report.h>

namespace xrtc
{
    namespace
    {
        const uint32_t kDefaultVideoSsrc = 1;
        const uint32_t kDefaultAudioSsrc = 0xFa17Fa17u;
    } // namespace
    struct SsrcInfo
    {
        uint32_t ssrc_id;
        std::string cname;
        std::string stream_id;
        std::string track_id;
    };

    static RtpDirection GetDirection(bool send, bool recv)
    {
        if (send && recv)
        {

            return RtpDirection::kSendRecv;
        }
        else if (send && !recv)
        {

            return RtpDirection::kSendOnly;
        }
        else if (!send && recv)
        {

            return RtpDirection::kRecvOnly;
        }
        else
        {
            return RtpDirection::kInactive;
        }
    }
    PeerConnection::PeerConnection(EventLoop *el, PortAllocator *allocator) : el_(el), clock_(webrtc::Clock::GetRealTimeClock()),
                                                                              transport_controller_(new TransportController(el, allocator))
    {
        transport_controller_->SignalCandidateAllocateDone.connect(this, &PeerConnection::OnCandidateAllocateDone);
        transport_controller_->SignalConnectionState.connect(this, &PeerConnection::OnConnectionState);

        transport_controller_->SignalRtpPacketReceived.connect(this, &PeerConnection::OnRtpPacketReceived);
        transport_controller_->SignalRtcpPacketReceived.connect(this, &PeerConnection::OnRtcpPacketReceived);
    }
    void PeerConnection::OnRtcpPacketReceived(TransportController *, rtc::CopyOnWriteBuffer *packet, int64_t ts)
    {
        if (IsLive())
        {
            if (audio_receive_stream_)
            {
                audio_receive_stream_->DeliverRtcp(packet->data(), packet->size());
            }
            if (video_receive_stream_)
            {
                video_receive_stream_->DeliverRtcp(packet->data(), packet->size());
            }
            if (video_send_stream_)
            {
                video_send_stream_->DeliverRtcp(packet->data(), packet->size());
            }
        }
        else
        {
            SignalRtcpPacketReceived(this, packet, ts);
        }
    }
    void PeerConnection::OnRtpPacketReceived(TransportController *, rtc::CopyOnWriteBuffer *packet, int64_t ts)
    {
        if (IsLive())
        {
            webrtc::RtpPacketReceived parsed_packet;
            if (!parsed_packet.Parse(std::move(*packet)))
            {
                RTC_LOG(LS_WARNING) << "invalid rt packet";
                return;
            }
            if (ts > 0)
            {
                parsed_packet.set_arrival_time(webrtc::Timestamp::Micros(ts));
            }
            else
            {
                parsed_packet.set_arrival_time(clock_->CurrentTime());
            }

            webrtc::MediaType packet_type = GetMediaType(parsed_packet.Ssrc());
            if (packet_type == webrtc::MediaType::VIDEO)
            {
                parsed_packet.set_payload_type_frequency(webrtc::kVideoPayloadTypeFrequency);
                if (video_receive_stream_)
                {
                    video_receive_stream_->OnRtpPacket(parsed_packet);
                }
            }
            else if (packet_type == webrtc::MediaType::AUDIO)
            {
                parsed_packet.set_payload_type_frequency(audio_clockrate_);
                if (audio_receive_stream_)
                {
                    audio_receive_stream_->OnRtpPacket(parsed_packet);
                }
            }
        }
        else
        {
            SignalRtpPacketReceived(this, packet, ts); // xrtc1.0
        }
    }
    void PeerConnection::OnConnectionState(TransportController *, PeerConnectionState state)
    {
        SignalConnectionState(this, state);
    }
    webrtc::MediaType PeerConnection::GetMediaType(uint32_t ssrc) const
    {
        if (ssrc == remote_video_ssrc_ || ssrc == remote_video_rtx_ssrc_)
        {
            return webrtc::MediaType::VIDEO;
        }
        else if (ssrc == remote_audio_ssrc_)
        {
            return webrtc::MediaType::AUDIO;
        }
        return webrtc::MediaType::ANY;
    }

    void DestoryTimerCb(EventLoop * /*el*/, TimerWatcher * /*w*/, void *data)
    {
        RTC_LOG(LS_INFO) << "DestoryTimerCb.";
        PeerConnection *pc = (PeerConnection *)data;
        delete pc;
    }
    void PeerConnection::Destory()
    {

        if (destory_timer_)
        {
            el_->DeleteTimer(destory_timer_);
            destory_timer_ = nullptr;
        }
        destory_timer_ = el_->CreateTimer(DestoryTimerCb, this, false);
        el_->StartTimer(destory_timer_, 10000); // 10ms
    }
    PeerConnection::~PeerConnection()
    {
        if (destory_timer_)
        {
            el_->DeleteTimer(destory_timer_);
            destory_timer_ = nullptr;
        }
        RTC_LOG(LS_INFO) << "PeerConnection destory.";
    }
    void PeerConnection::OnCandidateAllocateDone(TransportController * /*transport_controller*/, const std::string &transport_name,
                                                 IceCandidateComponent /*component*/, const std::vector<Candidate> &candidates)

    {

        for (auto c : candidates)
        {

            RTC_LOG(LS_INFO) << ".candidate gathering ,transport_name:" << transport_name << ", " << c.ToString();
        }
        if (!local_desc_)
        {
            return;
        }
        auto content = local_desc_->GetContent(transport_name);
        if (content)
        {
            content->AddCandidates(candidates);
        }
    }
    int PeerConnection::Init(rtc::RTCCertificate *certificate)
    {

        if (certificate)
        {
            certificate_ = certificate;
            transport_controller_->SetLocalCertificate(certificate);
        }
        else
        {
            // 不开启dtls

            is_dtls_ = false;
            transport_controller_->SetDtls(is_dtls_);
        }

        return 0;
    }

    std::string PeerConnection::CreateOffer(const RTCOfferAnswerOptions &options)
    {
        // 如果开启dtls 必须需要证书
        if (is_dtls_ && !certificate_)
        {
            RTC_LOG(LS_WARNING) << "certificate is null";
            return "";
        }
        local_desc_ = std::make_unique<SessionDescription>(SdpType::kOffer);
        // 随机生成ice
        IceParameters ice_param = IceCredentials::CreateRandomIceCredentials();
        // push stream              pull stream
        if (options.recv_audio || options.send_audio)
        {

            auto audio = std::make_shared<AudioContentDescription>();
            audio->SetDirection(GetDirection(options.send_audio, options.recv_audio));
            audio->SetRtcpMux(options.use_rtcp_mux);
            local_desc_->AddContent(audio);
            local_desc_->AddTransportInfo(audio->Mid(), ice_param, certificate_);
            if (options.send_audio)
            {
                for (auto stream : audio_source_)
                {
                    audio->AddStream(stream);
                }
                CreateAudioSendStream(audio.get());
            }
        }
        // push stream              pull stream
        if (options.recv_video || options.send_video)
        {
            auto video = std::make_shared<VideoContentDescription>();
            video->SetDirection(GetDirection(options.send_video, options.recv_video));
            video->SetRtcpMux(options.use_rtcp_mux);
            local_desc_->AddContent(video);

            local_desc_->AddTransportInfo(video->Mid(), ice_param, certificate_);

            if (options.send_video)
            {
                for (auto stream : video_source_)
                {
                    video->AddStream(stream);
                }
                CreateVideoSendStream(video.get());
            }
        }
        if (options.use_rtp_mux)
        {
            ContentGroup offer_bundle("BUNDLE");
            for (auto content : local_desc_->Contents())
            {
                offer_bundle.AddContentName(content->Mid());
            }
            if (!offer_bundle.ContentNames().empty())
            {
                local_desc_->AddGroup(offer_bundle);
            }
        }

        transport_controller_->SetLocalDescription(local_desc_.get());

        return local_desc_->ToString();
    }
    static std::string GetAttribute(const std::string &line)
    {

        // a=ice-ufrag:53oA\r\n
        std::vector<std::string> fileds;
        size_t size = rtc::tokenize(line, ':', &fileds);
        if (size != 2)
        {
            RTC_LOG(LS_WARNING) << "get attribute error:" << line;
            return "";
        }

        return fileds[1];
    }
    static int ParseTransportInfo(TransportDescription *td, const std::string &line)
    {
        if (line.find("a=ice-ufrag") != std::string::npos)
        {
            // a=ice-ufrag:NM9E
            td->ice_ufrag = GetAttribute(line);
            if (td->ice_ufrag.empty())
            {
                RTC_LOG(LS_WARNING) << "ice_ufrag is empty!";
                return -1;
            }
        }
        else if (line.find("a=ice-pwd") != std::string::npos)
        {
            // a=ice-pwd:iWFshiI+GCABgPbdsaT5Fzjk

            td->ice_pwd = GetAttribute(line);
            if (td->ice_pwd.empty())
            {
                RTC_LOG(LS_WARNING) << "ice_pwd is empty!";
                return -1;
            }
        }
        else if (line.find("a=fingerprint") != std::string::npos)
        {

            // a=fingerprint:sha-256 E3:C0:9A:29:C3:19:0C:AB:E9:28:EB:05:2D:FF:8A:21:7B:F6:C7:70:8D:78:28:45:E0:4F:20:3C:92:25:4B:E5
            std::vector<std::string> items;
            rtc::tokenize(line, ' ', &items);
            if (items.size() != 2)
            {
                RTC_LOG(LS_INFO) << "parse a=fingerprintf error: " << line;
                return -1;
            }
            // 获取算法
            std::string alg = items[0].substr(14);
            absl::c_transform(alg, alg.begin(), ::tolower); // 全部转化为小写字母
            std::string content = items[1];

            td->identify_fingerprint = rtc::SSLFingerprint::CreateUniqueFromRfc4572(alg, content);
            if (!(td->identify_fingerprint.get()))
            {
                RTC_LOG(LS_WARNING) << "create fingerpinrt error!" << line;
                return -1;
            }
        }
        return 0;
    }
    static int ParseSsrcGroupInfo(std::vector<SsrcGroup> &ssrc_groups, const std::string &line)
    {
        if (line.find("a=ssrc-group:") == std::string::npos)
        {
            return 0;
        }
        // rfc5576
        // a=ssrc-group:<semantics> <ssrc-id> ....
        std::vector<std::string> fields;
        rtc::split(line.substr(2), ' ', &fields);
        if (fields.size() < 2)
        {
            RTC_LOG(LS_WARNING) << "ssrc-group field size <2 ,line: " << line;
            return -1;
        }
        std::string semantics = GetAttribute(fields[0]);
        if (semantics.empty())
        {
            return -1;
        }
        std::vector<uint32_t> ssrcs;
        for (size_t i = 1; i < fields.size(); ++i)
        {
            uint32_t ssrc_id = 0;
            if (!rtc::FromString(fields[i], &ssrc_id))
            {
                return -1;
            }
            ssrcs.push_back(ssrc_id);
        }
        ssrc_groups.push_back(SsrcGroup(semantics, ssrcs));
        return 0;
    }
    static int ParseSsrcInfo(std::vector<SsrcInfo> &ssrc_info, const std::string &line)
    {
        if (line.find("a=ssrc:") == std::string::npos)
        {
            return 0;
        }
        // rfc5576
        // 1.a=ssrc:<ssrc-id> <attribute>
        // 2.a=ssrc::<ssrc-id> <attribute>:<value>
        std::string filed1, filed2;
        if (!rtc::tokenize_first(line.substr(2), ' ', &filed1, &filed2))
        {
            RTC_LOG(LS_WARNING) << "parse a=ssrc  failed ,line: " << line;
            return -1;
        }
        // a=ssrc:<ssrc-id>
        std::string ssrc_id_s = filed1.substr(5);
        uint32_t ssrc_id = 0;
        if (!rtc::FromString(ssrc_id_s, &ssrc_id))
        {
            RTC_LOG(LS_WARNING) << "invalid ssrc_id, line: " << line;
            return -1;
        }
        // attribute
        std::string attribute;
        std::string value;
        if (!rtc::tokenize_first(filed2, ':', &attribute, &value))
        {
            RTC_LOG(LS_WARNING) << "get ssrc attribute failed ,line: " << line;
            return -1;
        }

        auto iter = ssrc_info.begin();
        for (; iter != ssrc_info.end(); ++iter)
        {
            if (iter->ssrc_id == ssrc_id)
            {
                break;
            }
        }
        if (iter == ssrc_info.end())
        {
            SsrcInfo info;
            info.ssrc_id = ssrc_id;
            ssrc_info.push_back(info);
            iter = ssrc_info.end() - 1;
        }
        if ("cname" == attribute)
        {

            iter->cname = value;
        }
        else if ("msid" == attribute)
        {
            std::vector<std::string> fields;
            rtc::split(value, ' ', &fields);
            if (fields.size() < 1 || fields.size() > 2)
            {
                RTC_LOG(LS_WARNING) << "msid formate error line: " << line;
                return -1;
            }
            iter->stream_id = fields[0];
            if (fields.size() == 2)
            {
                iter->track_id = fields[1];
            }
        }
        return 0;
    }

    static void CreateTrackFormSsrcInfo(const std::vector<SsrcInfo> &ssrc_infos,
                                        std::vector<StreamParams> &tracks)
    {
        for (auto ssrc_info : ssrc_infos)
        {
            std::string track_id = ssrc_info.track_id;
            auto iter = tracks.begin();
            for (; iter != tracks.end(); ++iter)
            {
                if (iter->id == track_id)
                {
                    break;
                }
            }
            if (iter == tracks.end())
            {
                StreamParams track;
                track.id = track_id;
                tracks.push_back(track);
                iter = tracks.end() - 1;
            }
            iter->cname = ssrc_info.cname;
            iter->stream_id = ssrc_info.stream_id;
            iter->ssrcs.push_back(ssrc_info.ssrc_id);
        }
    }
    int PeerConnection::SetRemoteSdp(const std::string &sdp)
    {

        RTC_LOG(LS_INFO) << "pc set remote sdp!";
        // 解析answer sdp 信息
        std::vector<std::string> fields;
        size_t size = rtc::tokenize(sdp, '\n', &fields);
        if (size <= 0)
        {
            RTC_LOG(LS_WARNING) << "sdp invalid!";
            return -1;
        }
        bool is_rn = false;
        if (sdp.find("\r\n") != std::string::npos)
        {
            is_rn = true;
        }
        remote_desc_ = std::make_unique<SessionDescription>(SdpType::kAnswer);
        std::string media_type;

        auto audio_content = std::make_shared<AudioContentDescription>();
        auto video_content = std::make_shared<VideoContentDescription>();
        auto audio_td = std::make_shared<TransportDescription>();
        auto video_td = std::make_shared<TransportDescription>();
        std::vector<SsrcInfo> audio_ssrc_info;
        std::vector<SsrcInfo> video_ssrc_info;
        std::vector<SsrcGroup> video_ssrc_groups; // only video have attrite
        std::vector<StreamParams> audio_tracks;
        std::vector<StreamParams> video_tracks;

        for (auto field : fields)
        {
            if (is_rn)
            {
                field = field.substr(0, field.length() - 1);
            }
            // a=group:BUNDLE audio video
            if (field.find("a=group:BUNDLE") != std::string::npos)
            {
                std::vector<std::string> items;
                rtc::split(field, ' ', &items);
                if (items.size() > 1)
                {
                    ContentGroup answer_bundle("BUNDLE");
                    for (size_t i = 1; i < items.size(); ++i)
                    {
                        answer_bundle.AddContentName(items[i]);
                    }
                    remote_desc_->AddGroup(answer_bundle); //?????
                }
            }

            else if (field.find("m=") != std::string::npos)
            {
                std::vector<std::string> items;
                rtc::split(field, ' ', &items);
                if (items.size() <= 2)
                {
                    RTC_LOG(LS_WARNING) << "parse m = error: " << field;
                    return -1;
                }

                // m=audio 9 UDP/TLS/RTP/SAVPF 111
                media_type = items[0].substr(2);
                // RTC_LOG(LS_INFO) << "====media_type: " << media_type;
                if ("audio" == media_type)
                {
                    remote_desc_->AddContent(audio_content);
                    audio_td->mid = "audio";
                }
                else if ("video" == media_type)
                {
                    remote_desc_->AddContent(video_content);
                    video_td->mid = "video";
                }
            }
            if ("audio" == media_type)
            {
                if (ParseTransportInfo(audio_td.get(), field) != 0)
                {
                    RTC_LOG(LS_WARNING) << "parse_transport_info audio error!";
                    return -1;
                }
                if (ParseSsrcInfo(audio_ssrc_info, field) != 0)
                {
                    RTC_LOG(LS_WARNING) << "parse_ssrc_info audio error!";

                    return -1;
                }
            }
            else if ("video" == media_type)
            {
                if (ParseTransportInfo(video_td.get(), field) != 0)
                {

                    RTC_LOG(LS_WARNING) << ".......parse_transport_info video error!";
                    return -1;
                }
                if (ParseSsrcGroupInfo(video_ssrc_groups, field) != 0)
                {
                    return -1;
                }
                if (ParseSsrcInfo(video_ssrc_info, field) != 0)
                {
                    RTC_LOG(LS_WARNING) << "parse_ssrc_info video error!";
                    return -1;
                }
            }
        }
        if (!audio_ssrc_info.empty())
        {
            CreateTrackFormSsrcInfo(audio_ssrc_info, audio_tracks);
            for (auto track : audio_tracks)
            {
                audio_content->AddStream(track);
            }
        }
        if (!video_ssrc_info.empty())
        {
            CreateTrackFormSsrcInfo(video_ssrc_info, video_tracks);

            for (auto ssrc_group : video_ssrc_groups)
            {
                if (ssrc_group.ssrcs.empty())
                {
                    continue;
                }
                uint32_t ssrc = ssrc_group.ssrcs.front();
                for (StreamParams &track : video_tracks)
                {
                    if (track.HasSsrc(ssrc))
                    {
                        track.ssrc_groups.push_back(ssrc_group);
                    }
                }
            }
            for (auto track : video_tracks)
            {
                video_content->AddStream(track);
            }
        }
        remote_desc_->AddTransportInfo(audio_td);
        remote_desc_->AddTransportInfo(video_td);
        if (IsLive())
        {
            CreateVideoReceiveStream(video_content.get());
            CreateAudioReceiveStream(audio_content.get());
        }
        transport_controller_->SetRemoteDescription(remote_desc_.get());

        return 0;
    }
    void PeerConnection::CreateAudioReceiveStream(AudioContentDescription *audio_content)
    {
        // 按照系统的推拉流原子设计原则，一个 peerconnection 只允许推一个或者拉一个音频
        for (auto send_stream : audio_content->Streams())
        {
            uint32_t ssrc = send_stream.FirstSsrc();
            if (ssrc != 0)
            {
                remote_audio_ssrc_ = ssrc;

                AudioReceiveStreamConfig config;
                config.el = el_;
                config.clock = clock_;
                config.rtp.local_ssrc = kDefaultAudioSsrc;
                config.rtp.remote_ssrc = remote_audio_ssrc_;
                for (auto codec : audio_content->GetCodec())
                {
                    audio_clockrate_ = codec->clockrate;
                    break;
                }

                config.rtp_rtcp_module_observer = this;
                audio_receive_stream_ = std::make_unique<AudioReceiveStream>(config);
            }
            break;
        }
    }
    void PeerConnection::CreateVideoReceiveStream(VideoContentDescription *video_content)
    {
        // 按照系统的推拉流原子设计原则，一个 peerconnection 只允许推一个或者拉一个视频;
        for (auto send_stream : video_content->Streams())
        {
            uint32_t ssrc = send_stream.FirstSsrc();
            if (ssrc != 0)
            {
                remote_video_ssrc_ = ssrc;
                if (send_stream.ssrcs.size() >= 2)
                {
                    remote_video_rtx_ssrc_ = send_stream.ssrcs[1];
                }
                VideoReceiveStreamConfig config;
                config.el = el_;
                config.clock = clock_;
                config.rtp.local_ssrc = kDefaultVideoSsrc;
                config.rtp.remote_ssrc = remote_video_ssrc_;
                config.rtp.rtx_ssrc = remote_video_rtx_ssrc_;
                config.rtp.enable_xr = true;
                for (auto codec : video_content->GetCodec()) // 差异
                {
                    auto it = codec->codec_param.find("apt");
                    if (it != codec->codec_param.end())
                    {
                        config.rtp.rtx_associated_payload_types[codec->id] = std::stoi(it->second);
                    }
                }

                config.rtp_rtcp_module_observer = this;
                if (is_pli_)
                {
                    config.request_pli_interval_ms = 2000; // 修改成配置文件待做
                }
                video_receive_stream_ = std::make_unique<VideoReceiveStream>(config);
            }
            break;
        }
    }
    void PeerConnection::CreateAudioSendStream(AudioContentDescription *audio_content)
    {
        for (auto send_stream : audio_content->Streams())
        {
            uint32_t ssrc = send_stream.FirstSsrc();
            if (ssrc != 0)
            {
                local_audio_ssrc_ = ssrc;

                AudioSendStreamConfig config;
                config.el = el_;
                config.clock = clock_;
                config.rtp_rtcp_module_observer = this;
                config.rtp.local_ssrc = local_audio_ssrc_;

                audio_send_stream_ = std::make_unique<AudioSendStream>(config);
            }
        }
    }
    void PeerConnection::CreateVideoSendStream(VideoContentDescription *video_content)
    {
        for (auto send_stream : video_content->Streams())
        {
            uint32_t ssrc = send_stream.FirstSsrc();
            if (ssrc != 0)
            {
                local_video_ssrc_ = ssrc;
                if (send_stream.ssrcs.size() >= 2)
                {
                    local_video_rtx_ssrc_ = send_stream.ssrcs[1];
                }
                int rtx_payload_type = 0;
                for (auto codec : video_content->GetCodec())
                {
                    if ("rtx" == codec->name)
                    {
                        rtx_payload_type = codec->id;
                        break;
                    }
                }

                VideoSendStreamConfig config;
                config.el = el_;
                config.clock = clock_;
                config.rtp_rtcp_module_observer = this;
                config.rtp.local_ssrc = local_video_ssrc_;
                config.rtp.local_rtx_ssrc = local_video_rtx_ssrc_;
                config.rtp.rtx.payload_type = rtx_payload_type;

                video_send_stream_ = std::make_unique<VideoSendStream>(config);
            }
        }
    }
    int PeerConnection::SendPacket(webrtc::MediaType media_type,
                                   const webrtc::RtpPacketToSend &packet)
    {
        int ret = 0;
        if (webrtc::MediaType::VIDEO == media_type)
        {
            if (!video_send_stream_)
            {
                return -1;
            }
            if (webrtc::RtpPacketMediaType::kRetransmission == packet.packet_type())
            {
                // 构造rtx 数据包
                std::unique_ptr<webrtc::RtpPacketToSend> rtx_packet =
                    video_send_stream_->BuildRtxPacket(packet);
                if (!rtx_packet)
                {
                    return -1;
                }
                ret = SendRtp((const char *)rtx_packet->data(), rtx_packet->size());
                video_send_stream_->UpdateRtpStat(clock_->TimeInMilliseconds(), *rtx_packet);
            }
            else
            {
                ret = SendRtp((const char *)packet.data(), packet.size());
                video_send_stream_->UpdateRtpStat(clock_->TimeInMilliseconds(), packet);
            }
        }
        else if (webrtc::MediaType::AUDIO == media_type)
        {
            if (!audio_send_stream_)
            {
                return -1;
            }
            ret = SendRtp((const char *)packet.data(), packet.size());
            audio_send_stream_->UpdateRtpStat(clock_->TimeInMilliseconds(), packet);
        }
        return ret;
    }

    int PeerConnection::SendRtp(const char *data, size_t len)
    {
        if (transport_controller_)
        {
            return transport_controller_->SendRtp("audio", data, len);
        }
        return -1;
    }
    int PeerConnection::SendRtcp(const char *data, size_t len)
    {
        if (transport_controller_)
        {
            return transport_controller_->SendRtcp("audio", data, len);
        }
        return -1;
    }
    static void DebugCompoundRtcpPacket(const uint8_t *data, size_t len)
    {
        auto packet = rtc::MakeArrayView<const uint8_t>(data, len);
        webrtc::rtcp::CommonHeader rtcp_block;
        for (const uint8_t *next_block = packet.begin(); next_block != packet.end();
             next_block = rtcp_block.NextPacket())
        {
            ptrdiff_t remaining_block_size = packet.end() - next_block;
            if (!rtcp_block.Parse(next_block, remaining_block_size))
            {
                if (next_block == packet.begin())
                {
                    RTC_LOG(LS_WARNING) << "invalid incoming rtcp packet";
                    return;
                }
                break;
            }

            switch (rtcp_block.type())
            {

            case webrtc::rtcp::ReceiverReport::kPacketType:
            {
                webrtc::rtcp::ReceiverReport rr;
                if (rr.Parse(rtcp_block))
                {
                    if (rr.report_blocks().size() > 0)
                    {
                        for (auto report_block : rr.report_blocks())
                        {
                            RTC_LOG(LS_INFO) << "sourc_ssrc:" << report_block.source_ssrc()
                                             << ",fraction_lost:" << report_block.fraction_lost()
                                             << ",cumulative_lost:" << report_block.cumulative_lost_signed()
                                             << ",ext_high_seq_num:" << report_block.extended_high_seq_num()
                                             << ",jitter:" << report_block.jitter()
                                             << ",last_Sr:" << report_block.last_sr()
                                             << ",delay_since_last_sr:" << report_block.delay_since_last_sr();
                        }
                    }
                }
            }
            break;
            case webrtc::rtcp::SenderReport::kPacketType:
            {
                webrtc::rtcp::SenderReport sr;
                if (sr.Parse(rtcp_block))
                {
                    RTC_LOG(LS_INFO) << ".......local sr sender_ssrc:"
                                     << sr.sender_ssrc()
                                     << ",rtp_timestamp: " << sr.rtp_timestamp()
                                     << ",ntp: " << sr.ntp().ToMs()
                                     << ",packets count :" << sr.sender_packet_count()
                                     << ",octets count :" << sr.sender_octet_count();
                }
            }
            break;
            default:
                RTC_LOG(LS_WARNING) << "unknown rtcp packet_type :" << rtcp_block.type();
                break;
            }
        }
        return;
    }
    void PeerConnection::OnLocalRtcpPacket(webrtc::MediaType media_type, const uint8_t *data, size_t len)
    {
        // RTC_LOG(LS_WARNING) << "====on local rtcp packet ,len :" << len;
        DebugCompoundRtcpPacket(data, len);
        // 将本地打包好的的rtcp包，发送给对方
        SendRtcp((const char *)data, len);
    }
    void PeerConnection::OnFrame(std::unique_ptr<RtpFrameObject> frame)
    {
        // RTC_LOG(LS_WARNING) << "====>new frame ,frame_type:" << frame->frame_type()
        //                     << " ,[" << frame->first_seq_num() << ", " << frame->last_seq_num() << "]";
        SignalFrame(this, frame.get());
    }
    void PeerConnection::set_mode(const std::string &mode)
    {
        if ("transparent" == mode)
        {
            transport_mode_ = TransportMode::kTransparent;
        }
        else
        {
            transport_mode_ = TransportMode::kLive;
        }
    }
    void PeerConnection::OnRtpPacket(webrtc::MediaType media_type, const webrtc::RtpPacketReceived &packet)
    {
        SignalRtpPacket(this, media_type, packet);
    }
    void PeerConnection::OnSrInfo(webrtc::MediaType media_type, uint32_t rtp_timestamp, webrtc::NtpTime ntp)
    {
        SignalSrInfo(this, media_type, rtp_timestamp, ntp);
    }
    void PeerConnection::SetSrInfo(webrtc::MediaType media_type, uint32_t rtp_timestamp, webrtc::NtpTime ntp)
    {
        if (webrtc::MediaType::VIDEO == media_type)
        {
            if (video_send_stream_)
            {
                video_send_stream_->SetSrInfo(rtp_timestamp, ntp);
            }
        }
        else if (webrtc::MediaType::AUDIO == media_type)
        {

            if (audio_send_stream_)
            {
                audio_send_stream_->SetSrInfo(rtp_timestamp, ntp);
            }
        }
    }
    void PeerConnection::OnNackReceived(webrtc::MediaType media_type, const std::vector<uint16_t> &nack_list)
    {
        SignalNackReceived(this, media_type, nack_list);
    }
    static void RemoveOldReport(int64_t now, std::list<RttTime> *reports)
    {
        static const int64_t kRttTimeoutMs = 1500;
        reports->remove_if([&now](RttTime &item)
                           { return now - item.time > kRttTimeoutMs; });
    }
    static int64_t GetMaxRttTimeMs(const std::list<RttTime> &reports)
    {
        int64_t max_rtt_time_ms = -1;
        for (auto item : reports)
        {
            if (item.rtt_ms > max_rtt_time_ms)
            {
                max_rtt_time_ms = item.rtt_ms;
            }
        }
        return max_rtt_time_ms;
    }
    void PeerConnection::OnRttUpdate(int64_t rtt_ms)
    {
        // RTC_LOG(LS_WARNING) << "====rtt_ms:" << rtt_ms;
        int64_t now = clock_->TimeInMilliseconds();
        rtt_reports_.push_back(RttTime(now, rtt_ms));
        RemoveOldReport(now, &rtt_reports_);
        int64_t max_rtt_time_ms = GetMaxRttTimeMs(rtt_reports_);
        if (video_receive_stream_)
        {
            video_receive_stream_->OnRttUpdate(max_rtt_time_ms);
        }
    }
} // namespace xrtc